package com.sf.osms.dep.processor.nio;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import sun.nio.ch.FileChannelImpl;

import com.sf.osms.dep.processor.util.ExceptionUtils;


public class FileReader<T> implements Iterable<List<T>>{
	
	  private static Logger logger = LoggerFactory.getLogger(FileReader.class);
	  private static final long CHUNK_UNIT = 1024*1024;
	  
	  /**
	   * default chunk size is 1 M
	   */
	  private int size = 1;
	  private final Decoder<T> decoder;
	  private Iterator<File> files;
	 
	  private FileReader(Decoder<T> decoder, File... files) {
	    this(decoder, Arrays.asList(files));
	  }
	  private FileReader(Decoder<T> decoder, List<File> files) {
	    this.files = files.iterator();
	    this.decoder = decoder;
	  }
	  
	  public static <T> FileReader<T> create(Decoder<T> decoder, List<File> files) {
	    return new FileReader<T>(decoder, files);
	  }
	 
	  public static <T> FileReader<T> create(Decoder<T> decoder, File... files) {
	    return new FileReader<T>(decoder, files);
	  }
	  
	  /**
	   * set chunk size(unit is Mb)
	   * @param size
	   */
	  public void allocate(int size){
		  if(size > 0){
			  this.size = size;
		  }
	  }
	  
	  @Override
	  public Iterator<List<T>> iterator() {
	    return new Iterator<List<T>>() {
	      private List<T> entries;
	      private long chunkPos = 0;
	      private MappedByteBuffer buffer;
	      private FileChannel channel;
	      @Override
	      public boolean hasNext() {
	        if (buffer == null || !buffer.hasRemaining()) {
	          buffer = nextBuffer(chunkPos);
	          if (buffer == null) {
	            return false;
	          }
	        }
	        T result = null;
	        while ((result = decoder.decode(buffer)) != null) {
	          if (entries == null) {
	            entries = new ArrayList<T>();
	          }
	          entries.add(result);
	        }
	        // set next MappedByteBuffer chunk
	        chunkPos += buffer.position();
	      
	       
	        if (entries != null) {
	        	 unmap(buffer);
	        	 buffer = null;
	          return true;
	        } else {
	        	  try {
	  	        	unmap(buffer);
	  				channel.close();
	  				
	  			} catch (IOException e) {
	  				logger.error(ExceptionUtils.getExceptionStackString(e));
	  			}
	          buffer = null;
	          return false;
	        }
	      }
	 
	      private MappedByteBuffer nextBuffer(long position) {
	        try {
	          if (channel == null || channel.size() == position) {
	            if (channel != null) {
	            	 try {
	 	  	        	unmap(buffer);
	 	  				channel.close();
	 	  				
	 	  			} catch (IOException e) {
	 	  				logger.error(ExceptionUtils.getExceptionStackString(e));
	 	  			}
	              channel = null;
	            }
	            if (files.hasNext()) {
	              File file = files.next();
	              channel = new RandomAccessFile(file, "r").getChannel();
	              chunkPos = 0;
	              position = 0;
	            } else {
	              return null;
	            }
	          }
	          long chunkSize = size * CHUNK_UNIT;
	          if (channel.size() - position < chunkSize) {
	            chunkSize = channel.size() - position;
	          }
	           return channel.map(FileChannel.MapMode.READ_ONLY, chunkPos, chunkSize);
	        } catch (IOException e) {
	        	try {
	        		unmap(buffer);
					channel.close();
				} catch (IOException ex) {
					logger.error(ExceptionUtils.getExceptionStackString(ex));
				}
	        	
	        	throw new RuntimeException(e);
	        }
	      }
	 
	      @Override
	      public List<T> next() {
	        List<T> res = entries;
	        entries = null;
	        return res;
	      }
	 
	      @Override
	      public void remove() {
	        throw new UnsupportedOperationException();
	      }
	    };
	  }
	  
	  
	 public static void unmap(final MappedByteBuffer mappedByteBuffer) {
			try {
				if (mappedByteBuffer == null) {
					return;
				}
				
				  Method m = FileChannelImpl.class.getDeclaredMethod("unmap", MappedByteBuffer.class);  
			      m.setAccessible(true);  
			      m.invoke(FileChannelImpl.class, mappedByteBuffer); 
			} catch (Exception e) {
				logger.error(ExceptionUtils.getExceptionStackString(e));
			}
		}

}
